package com.atguigu.bigdata.spark.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object SparkStreaming01_WordCount {

    def main(args: Array[String]): Unit = {

        // TODO 创建环境对象

        // StreamingContext创建时，需要传递两个参数

        // 第一个参数表示环境配置
        //1.初始化 Spark 配置信息
        val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
        //2.初始化 SparkStreamingContext    第二个参数表示批量处理的周期（采集周期）
        val ssc = new StreamingContext(sparkConf, Seconds(3))


        // TODO 逻辑处理

        // 3.通过监控端口创建 DStream，读进来的数据为一行行   获取端口数据
        val lines: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)

        //将每一行数据做切分， 形成一个个单词
        val words = lines.flatMap(_.split(" "))

        //将单词映射成元组(word,1)
        val wordToOne = words.map((_, 1))

        //将相同的单词次数做统计
        val wordToCount: DStream[(String, Int)] = wordToOne.reduceByKey(_ + _)

        //打印
        wordToCount.print()


        // 由于SparkStreaming采集器是长期执行的任务，所以不能直接关闭
        // 如果main方法执行完毕，应用程序也会自动结束。所以不能让main执行完毕
        //ssc.stop()

        //启动 SparkStreamingContext
        // 1. 启动采集器
        ssc.start()
        // 2. 等待采集器的关闭
        ssc.awaitTermination()
    }
}

